// Kqueue 实现

#pragma once

#ifdef __APPLE__

#include <cstring>
#include <memory>
#include <thread>
#include <mutex>
#include <shared_mutex>
#include <atomic>
#include <unordered_map>

#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/event.h>

#include <hello/conc/conc_queue.h>
#include <hello/logger/log.h>


namespace hello
{

/**
 * 客户端连接信息存取处
 */
struct connection_stub
{
    std::unordered_map<int, sockaddr_in> _conns;

    std::shared_mutex _shm;

    ~connection_stub()
    {
        for (auto& [k, _] : _conns)
        {
            if (k > 0) close(k);
        }
    }

    inline void add_conn(int _fd, sockaddr_in _addr)
    {
        std::lock_guard<std::shared_mutex> lg(_shm);
        _conns.insert({ _fd, std::move(_addr) });
    }

    inline void remove_conn(int _fd)
    {
        std::lock_guard<std::shared_mutex> lg(_shm);
        _conns.erase(_fd);
    }

    inline bool exists(int _fd)
    {
        _shm.lock_shared();
        return _conns.find(_fd) != _conns.end();
        _shm.unlock_shared();
    }

    inline std::string get_conn_address(int _fd)
    {
        std::string ans;
        _shm.lock_shared();
        ans.append(inet_ntoa(_conns[_fd].sin_addr))
           .append(":")
           .append(std::to_string(ntohs(_conns[_fd].sin_port)));
        _shm.unlock_shared();
        return ans;
    }

};
connection_stub _conn_stub;


/**
 * 已读消息队列，等待task_executor处理
 */
struct in_msg_buffer
{
    hello::scqueue<std::string> _in_que;
    
    inline void push(const std::string& _msg)
    {
        _in_que.push(_msg);
    }

    inline std::string poll()
    {
        return _in_que.poll();
    }
};
in_msg_buffer _in_msg_buffer;


/**
 * 已处理消息容器，等待writer提取并写出
 */
struct out_msg_buffer
{
    std::shared_mutex _mut;

    std::unordered_map<int, std::vector<std::string>> _out_que;

    inline void insert(int _fd, std::string _msg)
    {
        std::lock_guard<std::shared_mutex> lg(_mut);
        _out_que[_fd].emplace_back(std::move(_msg));
    }

    inline std::string poll(int _fd)
    {
        std::lock_guard<std::shared_mutex> lg(_mut);
        std::string ans = _out_que[_fd].back();
        _out_que[_fd].pop_back();
        if (_out_que[_fd].empty())
        {
            _out_que.erase(_fd);
        }
        return ans;
    }

    inline bool has_write_task(int _fd)
    {
        bool ans = false;
        _mut.lock_shared();
        ans = (_out_que.find(_fd) != _out_que.end());
        _mut.unlock_shared();
        return ans;
    }

};
out_msg_buffer _out_msg_buffer;


void set_non_block(int _fd)
{
    fcntl(_fd, F_SETFL, fcntl(_fd, F_GETFL) | O_NONBLOCK);
}


/**
 * 从reactor，实现了事件监听的基本流程，具体的读写事件处理需要子类实现
 */
class worker
{
public:
    static constexpr size_t EVENT_CNT = 2048;
    static constexpr size_t BUF_SIZE = 2048;
    static constexpr size_t SIZE_T_LEN = sizeof(size_t);

public:
    worker() : _kqfd(-1), _shutdown(false)
    {
        if ((_kqfd = kqueue()) < 0)
        {
            logger::error("create listener kqueue error:", strerror(errno));
            throw std::runtime_error("create listener kqueue failed!");
        }
        set_non_block(_kqfd);
    }

    worker(const worker&) = delete;
    worker& operator = (const worker&) = delete;

    worker(worker&&) noexcept = delete;
    worker& operator = (worker&&) noexcept = delete;

public:
    virtual ~worker()
    {
        if (_pthread->joinable()) _pthread->join();
        shutdown();
        if (_kqfd >= 0) close(_kqfd);
    }

public:
    inline virtual void shutdown()
    {
        _shutdown.store(true);
    }

    inline virtual void remove_fd(int _fd)
    {
        _conn_stub.remove_conn(_fd);
        {
            struct kevent _remove_event
            {
                .ident = (uintptr_t)_fd,
                .filter = EVFILT_READ,
                .flags = EV_DELETE
            };
            std::lock_guard<std::mutex> lg(_mut);
            kevent(_kqfd, &_remove_event, 1, nullptr, 0, nullptr);
        }
    }

    inline virtual void close_fd(int _fd)
    {
        if (_fd > 0) close(_fd);
    }

public:
    inline virtual void start()
    {
        _pthread = std::make_shared<std::thread>([this]()
        {
            __process();
        });
    }

    inline virtual void __process()
    {
        _shutdown.store(false);
        struct kevent _events[EVENT_CNT];
        struct timespec _over_t { 0, 0 };
        while (!_shutdown.load())
        {
            int cnt = 0;
            {
                std::lock_guard<std::mutex> lg(_mut);
                cnt = kevent(_kqfd, nullptr, 0, _events, EVENT_CNT, &_over_t);
            }
            if (cnt < 0)
            {
                logger::error("kevent error:", strerror(errno));
                if (errno == EINTR) continue;
                else throw std::runtime_error("kevent failed!");
            }
            
            if (cnt > 0)
            {
                logger::debug("worker fetched", cnt, "events");
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
            }

            for (auto i = 0; i < cnt; ++i)
            {
                auto ifd = _events[i].ident;
                auto evts = _events[i].filter;
                if (evts == EVFILT_READ)
                {
                    __handle_read(ifd);
                }
                else if (evts == EVFILT_WRITE)
                {
                    __handle_write(ifd);
                }
                else if (evts == EVFILT_EXCEPT)
                {
                    logger::error("kqueue exception, fd:", ifd, "events:", evts);
                    remove_fd(ifd);
                    close_fd(ifd);
                }
                else
                {
                    logger::error("unexpected kqueue event, fd:", ifd, "events:", evts);
                    close_fd(ifd);
                }
            }
        }
    }

protected:
    inline virtual void __handle_read(int _fd) = 0;

    inline virtual void __handle_write(int _fd) = 0;

protected:
    int _kqfd;
    std::shared_ptr<std::thread> _pthread;
    std::atomic<bool> _shutdown;
    std::mutex _mut;

};


class reader : public worker
{
public:
    inline void register_ievent(int _fd)
    {
        struct kevent evt
        {
            .ident = (uintptr_t)_fd,
            .filter = EVFILT_READ,
            .flags = EV_ADD | EV_ENABLE | EV_CLEAR
        };
        logger::debug("begin register ievent");
        std::lock_guard<std::mutex> lg(_mut);
        logger::debug("begin kevent ev_read: add enable clear");
        kevent(_kqfd, &evt, 1, nullptr, 0, nullptr);
        logger::debug("after kevent ev_read add enable clear");
    }

public:
    inline void __handle_read(int _fd) override
    {
        logger::debug("begin handle read event:", _fd);
        if (_in_buf.find(_fd) == _in_buf.end())
        {
            _in_buf[_fd] = { -1, {} };
        }
        while (true)
        {
            size_t rest_len = -1;
            // 读取长度段
            if (_in_buf[_fd].second.size() < SIZE_T_LEN)
            {
                rest_len = SIZE_T_LEN - _in_buf[_fd].second.size();
                char rd_buf[rest_len];
                auto rcnt = read(_fd, rd_buf, rest_len);
                if (rcnt < 0)
                {
                    if (errno == EAGAIN || errno == EWOULDBLOCK)
                    {
                        logger::debug("read finished");
                        break;
                    }
                    else
                    {
                        logger::error("read error:", strerror(errno));
                        remove_fd(_fd);
                        close_fd(_fd);
                        return;
                    }
                }
                else if (rcnt == 0)
                {
                    logger::info("client closed:", _fd);
                    remove_fd(_fd);
                    close_fd(_fd);
                    return;
                }
                else
                {
                    _in_buf[_fd].second.append(rd_buf, rcnt);
                    if (_in_buf[_fd].second.size() >= SIZE_T_LEN)
                    {
                        _in_buf[_fd].first = *((size_t*)(_in_buf[_fd].second.c_str()));

                    }
                }
            }
            // 读取数据段
            else
            {
                rest_len = _in_buf[_fd].first - (_in_buf[_fd].second.size() - SIZE_T_LEN);
                char rd_buf[rest_len];
                auto rcnt = read(_fd, rd_buf, rest_len);
                if (rcnt < 0)
                {
                    if (errno == EAGAIN || errno == EWOULDBLOCK)
                    {
                        logger::debug("read finished");
                        break;
                    }
                    else
                    {
                        logger::error("read error:", strerror(errno));
                        _in_buf.erase(_fd);
                        remove_fd(_fd);
                        close_fd(_fd);
                        return;
                    }
                }
                else if (rcnt == 0)
                {
                    logger::info("client closed:", _fd);
                    _in_buf.erase(_fd);
                    remove_fd(_fd);
                    close_fd(_fd);
                    return;
                }
                else
                {
                    _in_buf[_fd].second.append(rd_buf, rcnt);
                    if (_in_buf[_fd].first == (_in_buf[_fd].second.size() - SIZE_T_LEN))
                    {
                        size_t tmp_fd = _fd;
                        memcpy(_in_buf[_fd].second.data(), &tmp_fd, SIZE_T_LEN);

                        _in_msg_buffer.push(_in_buf[_fd].second);
                        logger::debug("pushed read message to in_msg_buffer");

                        _in_buf[_fd].first = -1;
                        _in_buf[_fd].second = "";
                    }
                }
            }
        }
    }

    inline void __handle_write(int _fd) override {}

protected:
    std::unordered_map<int, std::pair<int, std::string>> _in_buf;

};


class writer : public worker
{
public:
    inline void register_oevent(int _fd)
    {
        logger::debug("begin register oevent", _fd);
        struct kevent evt
        {
            .ident = (uintptr_t)_fd,
            .filter = EVFILT_WRITE,
            .flags = EV_ADD | EV_ENABLE
        };
        std::lock_guard<std::mutex> lg(_mut);
        kevent(_kqfd, &evt, 1, nullptr, 0, nullptr);
        logger::debug("register oevent finished", _fd);
    }

    inline void remove_oevent(int _fd)
    {
        struct kevent evt
        {
            .ident = (uintptr_t)_fd,
            .filter = EVFILT_WRITE,
            .flags = EV_DELETE
        };
        logger::debug("begin remove oevent", _fd);
        std::lock_guard<std::mutex> lg(_mut);
        kevent(_kqfd, &evt, 1, nullptr, 0, nullptr);
        logger::debug("remove oevent finished", _fd);
    }

public:
    inline void __handle_write(int _fd) override
    {
        logger::debug("begin handle write event:", _fd);
        if (_out_buf.find(_fd) == _out_buf.end())
        {
            logger::debug("there was no writing task, begin poll from out_msg_buffer");
            std::string _msg = _out_msg_buffer.poll(_fd);
            logger::debug("polled a message from out_msg_buffer");
            _out_buf.insert({ _fd, { 0, std::move(_msg) } });
            logger::debug("insert the message into _out_buf");
        }
        auto offset = _out_buf[_fd].first;
        auto wcnt = write(_fd, _out_buf[_fd].second.c_str() + offset, _out_buf[_fd].second.size() - offset);
        _out_buf[_fd].first += wcnt;
        if (_out_buf[_fd].first == _out_buf[_fd].second.size())
        {
            _out_buf.erase(_fd);
            // 写完了，要注销写事件，并将其从写缓存中移除
            remove_oevent(_fd);
        }
        else
        {
            // 没写完，下次继续写，条件触发模式下不用重复注册
            // register_oevent(_fd);
        }
    }

    inline void __handle_read(int _fd) override {}

protected:
    std::unordered_map<int, std::pair<int, std::string>> _out_buf;
};


/**
 * 主reactor，负责接收新连接并注册到读reactor
 */
class listener
{
public:
    static constexpr size_t EVENT_CNT = 2048;
    static constexpr size_t BKLOG_SZ = 128;

public:
    listener(int _p, std::string _h, std::vector<std::shared_ptr<reader>> _pws) :
        _kqfd(-1),
        _lfd(-1),
        _port(_p),
        _host(std::move(_h)),
        _shutdown(false),
        _preaders(std::move(_pws))
    {
        if ((_kqfd = kqueue()) < 0)
        {
            logger::error("create listener kqueue error:", strerror(errno));
            throw std::runtime_error("create listener kqueue failed!");
        }
        set_non_block(_kqfd);

        if ((_lfd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
        {
            logger::error("create listener socket error:", strerror(errno));
            throw std::runtime_error("create listener socket failed!");
        }
        set_non_block(_lfd);

        sockaddr_in _saddr
        {
            .sin_family = AF_INET,
            .sin_port = htons(_port),
            .sin_addr = inet_addr(_host.c_str()),
            .sin_zero = { 0 }
        };
        if (bind(_lfd, (sockaddr*)&_saddr, sizeof(_saddr)) < 0)
        {
            logger::error("bind listener socket error:", strerror(errno));
            throw std::runtime_error("bind listener socket failed!");
        }

        if (listen(_lfd, BKLOG_SZ) < 0)
        {
            logger::error("listener socket listen error:", strerror(errno));
            throw std::runtime_error("listener socket listen failed!");
        }
        
        struct kevent _listen_events
        {
            .ident = (uintptr_t)_lfd,
            .filter = EVFILT_READ,
            .flags = EV_ADD | EV_ENABLE | EV_CLEAR
        };
        if (kevent(_kqfd, &_listen_events, 1, nullptr, 0, nullptr) < 0)
        {
            logger::error("register listener socket to listener kqueue error:", strerror(errno));
            throw std::runtime_error("egister listener socket to listener kqueue failed!");
        }
    }

    listener(const listener&) = delete;
    listener& operator = (const listener&) = delete;

    listener(listener&&) noexcept = delete;
    listener& operator = (listener&&) noexcept = delete;

public:
    ~listener()
    {
        if (_pthread->joinable()) _pthread->join();
        shutdown();
        if (_lfd >= 0) close(_lfd);
        if (_kqfd >= 0) close(_kqfd);
    }

public:
    inline void shutdown()
    {
        _shutdown.store(true);
    }

public:
    inline void start()
    {
        _pthread = std::make_shared<std::thread>([this]()
        {
            __process();
        });
    }

    inline void __process()
    {   
        struct kevent _events[EVENT_CNT];
        struct timespec _over_t { 0, 0 };
        while (!_shutdown.load())
        {
            auto cnt = kevent(_kqfd, nullptr, 0, _events, EVENT_CNT, &_over_t);
            if (cnt < 0)
            {
                logger::error("kevent error:", strerror(errno));
                if (errno == EINTR) continue;
                else throw std::runtime_error("kevent failed!");
            }
            if (cnt > 0) logger::debug("listener fetched", cnt, "events");

            for (auto i = 0; i < cnt; ++i)
            {
                auto ifd = _events[i].ident;
                auto evts = _events[i].filter;
                // 只监听读事件
                if (evts == EVFILT_READ)
                {
                    if (ifd == _lfd)
                    {
                        // 边缘触发模式下，要一直处理连接直到读缓冲空
                        while (true)
                        {
                            sockaddr_in _caddr {};
                            socklen_t _caddr_len = sizeof(_caddr);
                            auto _cfd = accept(_lfd, (sockaddr*)&_caddr, &_caddr_len);
                            if (_cfd < 0)
                            {
                                if (errno == EAGAIN || errno == EWOULDBLOCK)
                                    logger::debug("accept client conn finished!");
                                else
                                    logger::error("accept client conn error:", strerror(errno));
                                break;
                            }
                            else
                            {
                                //////// 将客户端添加到连接存储容器并将其FD注册到reader ////////
                                set_non_block(ifd);
                                auto _idx = rand() % (_preaders.size() - 1 - 0 + 1) + 0;
                                logger::debug("accept client:", _cfd, inet_ntoa(_caddr.sin_addr), ntohs(_caddr.sin_port));
                                _conn_stub.add_conn(_cfd, _caddr);
                                logger::debug("add it to conn_stub");
                                _preaders[_idx]->register_ievent(_cfd);
                                logger::debug("register it to reader");
                            }
                        }
                    }
                    else
                        logger::warn("unexpected fd for listener:", ifd, "listen fd:", _lfd);
                }
                else
                    logger::error("unexpected event for listener:", evts);
            }
        }
    }

private:
    int _kqfd;
    int _lfd;
    int _port;
    std::string _host;
    std::shared_ptr<std::thread> _pthread;
    std::atomic<bool> _shutdown;
    std::vector<std::shared_ptr<reader>> _preaders;

};


/**
 * 任务执行器，从in_msg_buffer中取消息处理
 */
class task_executor
{
public:
    static constexpr size_t SIZE_T_LEN = sizeof(size_t);

public:
    task_executor(size_t _conc,
                  std::function<std::string(const std::string&)> _handler,
                  std::vector<std::shared_ptr<writer>> _pws
    ) :
        _msg_handler(std::move(_handler)),
        _pwriters(std::move(_pws)),
        _shutdown(false)
    {
        _pthreads.resize(_conc);
    }

    task_executor(const task_executor&) = delete;
    task_executor& operator = (const task_executor&) = delete;

    task_executor(task_executor&&) = delete;
    task_executor& operator = (task_executor&&) = delete;

public:
    ~task_executor()
    {
        shutdown();
        for (auto& _pt : _pthreads)
        {
            if (_pt != nullptr)
            {
                if (_pt->joinable()) _pt->join();
            }
        }
    }

public:
    inline void start()
    {
        _shutdown.store(false);
        for (auto& _spt : _pthreads)
        {
            if (_spt == nullptr)
            {
                _spt = std::make_shared<std::thread>([this]()
                {
                    while (!_shutdown.load())
                    {
                        logger::debug("executor begin handle message");
                        std::string _msg = _in_msg_buffer.poll();
                        size_t _fd = *((size_t*)(_msg.c_str()));
                        _msg = _msg.substr(SIZE_T_LEN);

                        // 处理解析后的消息
                        std::string _ans = _msg_handler(_msg);

                        // 封装处理后的消息
                        size_t _ans_len = _ans.size();
                        std::string _out_msg;
                        _out_msg.append((char*)&_ans_len, SIZE_T_LEN);
                        _out_msg.append(_ans);

                        // 将封装后的消息添加到写消息容器
                        _out_msg_buffer.insert(_fd, _out_msg);

                        // 将客户端FD添加到writer并注册写事件
                        auto _idx = rand() % (_pwriters.size() - 1 - 0 + 1) + 0;
                        _pwriters[_idx]->register_oevent(_fd);
                        logger::debug("executor handle message finished");
                    }
                });
            }
        }
    }

    inline void shutdown()
    {
        _shutdown.store(true);
    }

public:
    std::function<std::string(const std::string&)> _msg_handler;
    std::vector<std::shared_ptr<std::thread>> _pthreads;
    std::vector<std::shared_ptr<writer>> _pwriters;
    std::atomic<bool> _shutdown;

};


class tcp_server
{
public:
    tcp_server(int _p,
               std::string _h,
               std::function<std::string(const std::string&)> _f
    ) :
        _port(_p),
        _host(std::move(_h)),
        _handler(std::move(_f)),
        _reader_cnt(1),
        _writer_cnt(1),
        _executor_cnt(1)
    {
    }

    tcp_server(int _p,
               std::string _h,
               std::function<std::string(const std::string&)> _f,
               size_t _reader_sz,
               size_t _writer_sz,
               size_t _executor_sz
    ) :
        _port(_p),
        _host(std::move(_h)),
        _handler(std::move(_f)),
        _reader_cnt(_reader_sz),
        _writer_cnt(_writer_sz),
        _executor_cnt(_executor_sz)
    {
    }

public:
    inline void start()
    {
        std::vector<std::shared_ptr<reader>> _spreaders;
        _spreaders.resize(_reader_cnt);
        for (auto& _spw : _spreaders)
        {
            _spw = std::make_shared<reader>();
            _spw->start();
        }

        logger::warn("readers started");

        std::vector<std::shared_ptr<writer>> _spwriters;
        _spwriters.resize(_writer_cnt);
        for (auto& _spw : _spwriters)
        {
            _spw = std::make_shared<writer>();
            _spw->start();
        }

        logger::warn("writer started");

        task_executor _te(_executor_cnt, _handler, _spwriters);
        _te.start();

        logger::warn("executor started");

        listener _lsn(_port, _host, _spreaders);
        _lsn.start();

        logger::warn("listener started");
    }

private:
    int _port;
    std::string _host;
    std::function<std::string(const std::string&)> _handler;
    size_t _reader_cnt;
    size_t _writer_cnt;
    size_t _executor_cnt;
};


}

#endif
